package com.sync.platform.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.sync.platform.entity.DatabaseConfig;
import com.sync.platform.entity.ServerConfig;
import com.sync.platform.entity.SyncTask;
import com.sync.platform.entity.SyncTaskLog;
import com.sync.platform.mapper.SyncTaskMapper;
import com.sync.platform.service.DatabaseConfigService;
import com.sync.platform.service.ServerConfigService;
import com.sync.platform.service.SyncTaskService;
import com.sync.platform.service.SyncTaskLogService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.stream.Collectors;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.ArrayNode;

import java.sql.Connection;
import java.sql.Statement;
import java.sql.ResultSet;
import java.sql.PreparedStatement;
import java.sql.SQLException;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.io.UnsupportedEncodingException;
import java.io.File;

/**
 * Sync task service implementation
 */
@Service
public class SyncTaskServiceImpl extends ServiceImpl<SyncTaskMapper, SyncTask> implements SyncTaskService {

    private static final Logger logger = LoggerFactory.getLogger(SyncTaskServiceImpl.class);
    
    private static final int TASK_TYPE_DATABASE = 1;
    private static final int TASK_TYPE_FILE = 2;
    
    private static final int STATUS_DISABLED = 0;
    private static final int STATUS_ENABLED = 1;
    
    private static final int EXECUTION_FAILED = 0;
    private static final int EXECUTION_SUCCESS = 1;
    
    @Autowired
    private DatabaseConfigService databaseConfigService;
    
    @Autowired
    private ServerConfigService serverConfigService;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    @Autowired
    private SyncTaskLogService syncTaskLogService;

    @Override
    public boolean saveDatabaseSyncTask(SyncTask syncTask, List<String> sourceTables, 
                                       boolean createTableIfNotExists, boolean truncateBeforeSync) {
        // Forward to the extended method with default values for incremental sync
        return saveDatabaseSyncTask(syncTask, sourceTables, createTableIfNotExists, truncateBeforeSync, false, null);
    }
    
    @Override
    public boolean saveDatabaseSyncTask(SyncTask syncTask, List<String> sourceTables, 
                                       boolean createTableIfNotExists, boolean truncateBeforeSync,
                                       boolean incrementalSync, String incrementalColumn) {
        try {
            // Set task type to database sync
            syncTask.setTaskType(TASK_TYPE_DATABASE);
            
            // Set status to enabled by default
            if (syncTask.getStatus() == null) {
                syncTask.setStatus(STATUS_ENABLED);
            }
            
            // Create JSON config
            ObjectNode config = objectMapper.createObjectNode();
            
            // Add selected tables
            if (sourceTables != null && !sourceTables.isEmpty()) {
                ArrayNode tablesNode = config.putArray("tables");
                for (String table : sourceTables) {
                    tablesNode.add(table);
                }
            }
            
            // Add sync options
            config.put("createTableIfNotExists", createTableIfNotExists);
            config.put("truncateBeforeSync", truncateBeforeSync);
            config.put("incrementalSync", incrementalSync);
            
            if (incrementalSync && incrementalColumn != null && !incrementalColumn.isEmpty()) {
                config.put("incrementalColumn", incrementalColumn);
            }
            
            // Save config as JSON string
            syncTask.setConfig(objectMapper.writeValueAsString(config));
            
            // Save task
            return saveOrUpdate(syncTask);
        } catch (Exception e) {
            logger.error("Error saving database sync task: {}", e.getMessage(), e);
            return false;
        }
    }

    @Override
    public List<SyncTask> getDatabaseSyncTasks() {
        LambdaQueryWrapper<SyncTask> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SyncTask::getTaskType, TASK_TYPE_DATABASE);
        wrapper.orderByDesc(SyncTask::getCreateTime);
        return list(wrapper);
    }

    @Override
    public List<SyncTask> getFileSyncTasks() {
        LambdaQueryWrapper<SyncTask> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SyncTask::getTaskType, TASK_TYPE_FILE);
        wrapper.orderByDesc(SyncTask::getCreateTime);
        return list(wrapper);
    }

    @Override
    public Map<String, Object> saveFileSyncTask(SyncTask syncTask, String sourcePath, String targetPath, String filePattern, 
                                              String transferType, String deleteSource, String syncDirectories) {
        Map<String, Object> result = new HashMap<>();
        result.put("success", false);
        
        try {
            // Set task type to file sync
            syncTask.setTaskType(TASK_TYPE_FILE);
            
            // Set status to enabled by default
            if (syncTask.getStatus() == null) {
                syncTask.setStatus(STATUS_ENABLED);
            }
            
            // Create config string with format: sourcePath|targetPath|filePattern|transferType|deleteSource|syncDirectories
            StringBuilder configBuilder = new StringBuilder();
            configBuilder.append(sourcePath).append("|").append(targetPath);
            
            if (filePattern != null && !filePattern.isEmpty()) {
                configBuilder.append("|").append(filePattern);
            } else {
                configBuilder.append("|");
            }
            
            // Add transfer type (ssh or ftp)
            configBuilder.append("|").append(transferType);
            
            // Add delete source files option
            configBuilder.append("|").append(deleteSource);
            
            // Add sync directories option
            configBuilder.append("|").append(syncDirectories);
            
            syncTask.setConfig(configBuilder.toString());
            
            // Save task
            boolean saveResult = saveOrUpdate(syncTask);
            
            if (saveResult) {
                result.put("success", true);
                result.put("message", "文件同步任务保存成功");
                result.put("taskId", syncTask.getId());
            } else {
                result.put("message", "保存任务失败");
            }
        } catch (Exception e) {
            logger.error("Error saving file sync task: {}", e.getMessage(), e);
            result.put("message", "保存任务出错: " + e.getMessage());
        }
        
        return result;
    }

    @Override
    public Map<String, Object> executeTask(Long id) {
        Map<String, Object> result = new HashMap<>();
        result.put("success", false);
        
        SyncTask task = getById(id);
        if (task == null) {
            result.put("message", "任务不存在");
            return result;
        }
        
        try {
            // Set execution time
            task.setLastExecutionTime(LocalDateTime.now());
            
            // Create log entry
            Long logId = null;
            if (task.getTaskType() == TASK_TYPE_DATABASE) {
                logId = syncTaskLogService.createLog(id, TASK_TYPE_DATABASE, "开始执行数据库同步任务", null);
            } else if (task.getTaskType() == TASK_TYPE_FILE) {
                logId = syncTaskLogService.createLog(id, TASK_TYPE_FILE, "开始执行文件同步任务", null);
            }
            
            if (task.getTaskType() == TASK_TYPE_DATABASE) {
                // Database sync task
                result = executeDatabaseSyncTask(task, logId);
            } else if (task.getTaskType() == TASK_TYPE_FILE) {
                // File sync task
                result = executeFileSyncTask(task, logId);
            } else {
                result.put("message", "未知的任务类型");
            }
            
            // Update task execution result
            boolean success = (boolean) result.get("success");
            task.setLastExecutionResult(success ? EXECUTION_SUCCESS : EXECUTION_FAILED);
            task.setLastExecutionMessage((String) result.get("message"));
            updateById(task);
            
            return result;
        } catch (Exception e) {
            logger.error("Error executing task {}: {}", id, e.getMessage(), e);
            
            // Update task execution result
            task.setLastExecutionResult(EXECUTION_FAILED);
            task.setLastExecutionMessage("执行出错: " + e.getMessage());
            updateById(task);
            
            result.put("message", "执行出错: " + e.getMessage());
            return result;
        }
    }
    
    private Map<String, Object> executeDatabaseSyncTask(SyncTask task, Long logId) {
        Map<String, Object> result = new HashMap<>();
        result.put("success", false);
        
        StringBuilder detailLog = new StringBuilder();
        
        try {
            // Get source and target databases
            DatabaseConfig sourceDb = databaseConfigService.getById(task.getSourceId());
            DatabaseConfig targetDb = databaseConfigService.getById(task.getTargetId());
            
            if (sourceDb == null) {
                String message = "源数据库不存在";
                result.put("message", message);
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                }
                return result;
            }
            
            if (targetDb == null) {
                String message = "目标数据库不存在";
                result.put("message", message);
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                }
                return result;
            }
            
            detailLog.append("源数据库: ").append(sourceDb.getName()).append(" (").append(sourceDb.getDbType()).append(")\n");
            detailLog.append("目标数据库: ").append(targetDb.getName()).append(" (").append(targetDb.getDbType()).append(")\n\n");
            
            // Parse configuration
            ObjectNode config = (ObjectNode) objectMapper.readTree(task.getConfig());
            
            // Get tables to sync
            List<String> tables;
            if (config.has("tables")) {
                ArrayNode tablesNode = (ArrayNode) config.get("tables");
                tables = new ArrayList<>();
                for (int i = 0; i < tablesNode.size(); i++) {
                    tables.add(tablesNode.get(i).asText());
                }
            } else {
                // Sync all tables if not specified
                tables = databaseConfigService.getTables(task.getSourceId());
            }
            
            if (tables.isEmpty()) {
                String message = "没有找到要同步的表";
                result.put("message", message);
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                }
                return result;
            }
            
            // Get sync options
            boolean createTableIfNotExists = config.has("createTableIfNotExists") && 
                                            config.get("createTableIfNotExists").asBoolean();
            boolean truncateBeforeSync = config.has("truncateBeforeSync") && 
                                        config.get("truncateBeforeSync").asBoolean();
                                        
            // Get incremental sync options
            boolean incrementalSync = config.has("incrementalSync") && 
                                     config.get("incrementalSync").asBoolean();
            String incrementalColumn = config.has("incrementalColumn") ? 
                                      config.get("incrementalColumn").asText() : null;
            
            // Get last execution time for incremental sync
            LocalDateTime lastSyncTime = task.getLastSuccessTime();
            
            logger.info("Starting to sync {} tables from {} to {}", tables.size(), sourceDb.getName(), targetDb.getName());
            detailLog.append("同步选项:\n");
            detailLog.append("- 表数量: ").append(tables.size()).append("\n");
            detailLog.append("- 表: ").append(String.join(", ", tables)).append("\n");
            detailLog.append("- 如表不存在则创建: ").append(createTableIfNotExists).append("\n");
            detailLog.append("- 同步前清空目标表: ").append(truncateBeforeSync).append("\n");
            detailLog.append("- 增量同步: ").append(incrementalSync).append("\n");
            
            if (incrementalSync) {
                detailLog.append("- 增量字段: ").append(incrementalColumn).append("\n");
                if (lastSyncTime != null) {
                    detailLog.append("- 上次同步时间: ").append(lastSyncTime).append("\n");
                } else {
                    detailLog.append("- 未找到上次同步时间，将执行全量同步\n");
                }
            }
            
            detailLog.append("\n开始同步...\n\n");
            
            logger.info("Tables: {}", String.join(", ", tables));
            logger.info("Options: createTableIfNotExists={}, truncateBeforeSync={}, incrementalSync={}, column={}",
                      createTableIfNotExists, truncateBeforeSync, incrementalSync, incrementalColumn);
            if (incrementalSync && lastSyncTime != null) {
                logger.info("Incremental sync from: {}", lastSyncTime);
            }
            
            // Create connections to source and target databases
            Connection sourceConn = null;
            Connection targetConn = null;
            try {
                // Connect to source database
                sourceConn = getConnection(sourceDb);
                if (sourceConn == null) {
                    String message = "无法连接到源数据库";
                    result.put("message", message);
                    detailLog.append("错误: ").append(message).append("\n");
                    if (logId != null) {
                        syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                    }
                    return result;
                }

                // Connect to target database
                targetConn = getConnection(targetDb);
                if (targetConn == null) {
                    String message = "无法连接到目标数据库";
                    result.put("message", message);
                    detailLog.append("错误: ").append(message).append("\n");
                    if (logId != null) {
                        syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                    }
                    return result;
                }
                
                detailLog.append("成功连接到源数据库和目标数据库\n\n");
                
                // Set auto-commit to false for batch operations
                targetConn.setAutoCommit(false);
                
                // Initialize counters for summary
                int successTables = 0;
                int totalRows = 0;
                Map<String, Integer> tableRowCounts = new HashMap<>();
                List<String> failedTables = new ArrayList<>();
                
                // For each table to sync
                for (String tableName : tables) {
                    try {
                        logger.info("Syncing table: {}", tableName);
                        detailLog.append("同步表 ").append(tableName).append("...\n");
                        
                        int rowCount = syncTable(
                            sourceConn, 
                            targetConn, 
                            tableName, 
                            createTableIfNotExists, 
                            truncateBeforeSync,
                            incrementalSync,
                            incrementalColumn,
                            lastSyncTime
                        );
                        
                        if (rowCount >= 0) {
                            successTables++;
                            totalRows += rowCount;
                            tableRowCounts.put(tableName, rowCount);
                            logger.info("Successfully synced {} rows for table {}", rowCount, tableName);
                            detailLog.append("✅ 成功同步 ").append(rowCount).append(" 行数据\n");
                        } else {
                            failedTables.add(tableName);
                            logger.error("Failed to sync table {}", tableName);
                            detailLog.append("❌ 同步失败\n");
                        }
                    } catch (Exception e) {
                        failedTables.add(tableName);
                        logger.error("Error syncing table {}: {}", tableName, e.getMessage(), e);
                        detailLog.append("❌ 同步出错: ").append(e.getMessage()).append("\n");
                    }
                }
                
                // Commit all changes
                targetConn.commit();
                
                detailLog.append("\n同步完成\n");
                detailLog.append("成功同步表数: ").append(successTables).append("/").append(tables.size()).append("\n");
                detailLog.append("总行数: ").append(totalRows).append("\n");
                
                if (!failedTables.isEmpty()) {
                    detailLog.append("失败表: ").append(String.join(", ", failedTables)).append("\n");
                }
                
                detailLog.append("\n表详情:\n");
                for (Map.Entry<String, Integer> entry : tableRowCounts.entrySet()) {
                    detailLog.append("- ").append(entry.getKey()).append(": ").append(entry.getValue()).append(" 行\n");
                }
                
                // If the sync was successful, update the last successful sync time
                if (failedTables.isEmpty() && totalRows > 0) {
                    task.setLastSuccessTime(LocalDateTime.now());
                    updateById(task);
                }
                
                // Build result
                boolean allSuccess = failedTables.isEmpty();
                result.put("success", allSuccess);
                result.put("tableCount", tables.size());
                result.put("successTables", successTables);
                result.put("failedTables", failedTables);
                result.put("totalRows", totalRows);
                result.put("tableRowCounts", tableRowCounts);
                result.put("tables", tables);
                result.put("incrementalSync", incrementalSync);
                
                String resultMessage;
                if (allSuccess) {
                    resultMessage = String.format("同步任务执行成功: 同步了 %d 个表，共 %d 行数据", successTables, totalRows);
                } else {
                    resultMessage = String.format("同步任务部分成功: 成功同步 %d 个表，失败 %d 个表", 
                            successTables, failedTables.size());
                }
                
                result.put("message", resultMessage);
                
                // Update log with final result
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, 
                                              allSuccess ? EXECUTION_SUCCESS : EXECUTION_FAILED, 
                                              totalRows, 
                                              resultMessage, 
                                              detailLog.toString());
                }
                
                return result;
            } finally {
                // Close connections
                if (sourceConn != null) {
                    try {
                        sourceConn.close();
                    } catch (SQLException e) {
                        logger.error("Error closing source database connection", e);
                    }
                }
                if (targetConn != null) {
                    try {
                        targetConn.setAutoCommit(true); // Reset auto-commit
                        targetConn.close();
                    } catch (SQLException e) {
                        logger.error("Error closing target database connection", e);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("Error executing database sync task: {}", e.getMessage(), e);
            String message = "同步任务执行失败: " + e.getMessage();
            result.put("message", message);
            
            // Add error to detail log
            detailLog.append("\n执行出错: ").append(e.getMessage()).append("\n");
            
            // Update log with error
            if (logId != null) {
                syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
            }
            
            return result;
        }
    }
    
    private Map<String, Object> executeFileSyncTask(SyncTask task, Long logId) {
        Map<String, Object> result = new HashMap<>();
        result.put("success", false);
        
        StringBuilder detailLog = new StringBuilder();
        detailLog.append("开始文件同步任务 [").append(task.getName()).append("]\n");
        
        try {
            // Parse config
            String config = task.getConfig();
            if (config == null || config.isEmpty()) {
                String message = "同步配置为空";
                result.put("message", message);
                detailLog.append(message).append("\n");
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                }
                return result;
            }
            
            String[] parts = config.split("\\|");
            if (parts.length < 2) {
                String message = "同步配置格式错误";
                result.put("message", message);
                detailLog.append(message).append("\n");
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                }
                return result;
            }
            
            String sourcePath = parts[0];
            String targetPath = parts[1];
            String filePattern = parts.length > 2 ? parts[2] : null;
            String transferType = parts.length > 3 ? parts[3] : "ssh"; // Default to SSH
            boolean deleteSourceFiles = parts.length > 4 && "true".equals(parts[4]);
            boolean syncDirectories = parts.length > 5 && "true".equals(parts[5]);
            
            detailLog.append("源路径: ").append(sourcePath).append("\n");
            detailLog.append("目标路径: ").append(targetPath).append("\n");
            if (filePattern != null && !filePattern.isEmpty()) {
                detailLog.append("文件过滤: ").append(filePattern).append("\n");
            }
            detailLog.append("传输方式: ").append(transferType).append("\n");
            detailLog.append("传输后删除源文件: ").append(deleteSourceFiles).append("\n");
            detailLog.append("同步目录结构: ").append(syncDirectories).append("\n");
            
            // Get source and target servers
            ServerConfig sourceServer = serverConfigService.getById(task.getSourceId());
            ServerConfig targetServer = serverConfigService.getById(task.getTargetId());
            
            if (sourceServer == null) {
                String message = "源服务器不存在";
                result.put("message", message);
                detailLog.append(message).append("\n");
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                }
                return result;
            }
            
            if (targetServer == null) {
                String message = "目标服务器不存在";
                result.put("message", message);
                detailLog.append(message).append("\n");
                if (logId != null) {
                    syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, message, detailLog.toString());
                }
                return result;
            }
            
            detailLog.append("源服务器: ").append(sourceServer.getName()).append(" (").append(sourceServer.getHost()).append(")\n");
            detailLog.append("目标服务器: ").append(targetServer.getName()).append(" (").append(targetServer.getHost()).append(")\n\n");
            
            // Execute transfer based on selected transfer type
            List<String> syncedFiles = new ArrayList<>();
            int syncCount = 0;
            
            if ("ftp".equals(transferType)) {
                // Use FTP for file transfer
                detailLog.append("使用FTP协议传输文件\n");
                
                // Connect to source FTP server
                org.apache.commons.net.ftp.FTPClient sourceFtp = null;
                try {
                    sourceFtp = com.sync.platform.util.FTPUtil.connect(
                        sourceServer.getHost(),
                        sourceServer.getPort(),
                        sourceServer.getUsername(),
                        sourceServer.getPassword()
                    );
                    
                    detailLog.append("成功连接到源FTP服务器\n");
                    
                    // Connect to target FTP server
                    org.apache.commons.net.ftp.FTPClient targetFtp = null;
                    try {
                        targetFtp = com.sync.platform.util.FTPUtil.connect(
                            targetServer.getHost(),
                            targetServer.getPort(),
                            targetServer.getUsername(),
                            targetServer.getPassword()
                        );
                        
                        detailLog.append("成功连接到目标FTP服务器\n");
                        
                        // List files on source server
                        org.apache.commons.net.ftp.FTPFile[] files = com.sync.platform.util.FTPUtil.listFiles(sourceFtp, sourcePath);
                        detailLog.append("源目录文件数: ").append(files.length).append("\n");
                        
                        // Create temporary directory for file transfer
                        String tempDir = System.getProperty("java.io.tmpdir") + "/ftp_sync_" + task.getId() + "_" + System.currentTimeMillis();
                        new File(tempDir).mkdirs();
                        detailLog.append("创建临时目录: ").append(tempDir).append("\n");
                        
                        // Process file patterns
                        String[] patterns = com.sync.platform.util.FilePatternUtil.parsePatterns(filePattern);
                        if (patterns != null) {
                            detailLog.append("文件过滤模式: ").append(filePattern).append("\n");
                            for (String pattern : patterns) {
                                detailLog.append("过滤模式: ").append(com.sync.platform.util.FilePatternUtil.getPatternDescription(pattern)).append("\n");
                            }
                        } else {
                            detailLog.append("没有设置文件过滤模式，将同步所有文件\n");
                        }
                        
                        // Download and upload files
                        for (org.apache.commons.net.ftp.FTPFile file : files) {
                            String fileName = file.getName();
                            boolean isDirectory = file.isDirectory();
                            
                            detailLog.append("处理项目: ").append(fileName).append(", 类型: ").append(isDirectory ? "目录" : "文件").append("\n");
                            
                            if (isDirectory) {
                                // Process directory - create it but don't filter directories by pattern
                                String sourceSubDir = sourcePath + (sourcePath.endsWith("/") ? "" : "/") + fileName;
                                String targetSubDir = targetPath + (targetPath.endsWith("/") ? "" : "/") + fileName;
                                
                                // Skip directory if not syncing directories
                                if (!syncDirectories) {
                                    detailLog.append("跳过目录 ").append(fileName).append(" (未启用目录同步)\n");
                                    continue;
                                }
                                
                                detailLog.append("创建目标子目录: ").append(targetSubDir).append("\n");
                                
                                try {
                                    // Create target directory
                                    com.sync.platform.util.FTPUtil.createRemoteDirectories(targetFtp, targetSubDir);
                                    syncedFiles.add(fileName + "/");
                                    syncCount++;
                                    detailLog.append("目录 ").append(fileName).append(" 同步成功\n");
                                } catch (Exception e) {
                                    detailLog.append("创建目标子目录失败: ").append(e.getMessage()).append("\n");
                                }
                            } else {
                                // Process file
                                // Check if file matches pattern
                                if (patterns != null) {
                                    boolean matched = com.sync.platform.util.FilePatternUtil.matchAnyPattern(fileName, patterns);
                                    if (!matched) {
                                        detailLog.append("文件 ").append(fileName).append(" 不匹配任何过滤模式，跳过\n");
                                        continue;
                                    } else {
                                        detailLog.append("文件 ").append(fileName).append(" 匹配过滤模式\n");
                                    }
                                }
                                
                                String sourceFilePath = sourcePath + (sourcePath.endsWith("/") ? "" : "/") + fileName;
                                String targetFilePath = targetPath + (targetPath.endsWith("/") ? "" : "/") + fileName;
                                String tempFilePath = tempDir + "/" + fileName;
                                
                                detailLog.append("源文件路径: ").append(sourceFilePath).append("\n");
                                detailLog.append("目标文件路径: ").append(targetFilePath).append("\n");
                                detailLog.append("临时文件路径: ").append(tempFilePath).append("\n");
                                
                                // Download file from source
                                boolean downloadSuccess = com.sync.platform.util.FTPUtil.downloadFile(sourceFtp, sourceFilePath, tempFilePath);
                                detailLog.append("下载文件结果: ").append(downloadSuccess ? "成功" : "失败").append("\n");
                                
                                if (downloadSuccess) {
                                    // Upload file to target
                                    boolean uploadSuccess = com.sync.platform.util.FTPUtil.uploadFile(targetFtp, tempFilePath, targetFilePath);
                                    detailLog.append("上传文件结果: ").append(uploadSuccess ? "成功" : "失败").append("\n");
                                    
                                    if (uploadSuccess) {
                                        syncedFiles.add(fileName);
                                        syncCount++;
                                        detailLog.append("文件 ").append(fileName).append(" 同步成功\n");
                                        
                                        // Delete source file if required
                                        if (deleteSourceFiles) {
                                            boolean deleteSuccess = sourceFtp.deleteFile(sourceFilePath);
                                            detailLog.append("删除源文件结果: ").append(deleteSuccess ? "成功" : "失败").append("\n");
                                        }
                                        
                                        // Delete temp file
                                        boolean tempDeleteSuccess = new File(tempFilePath).delete();
                                        detailLog.append("删除临时文件结果: ").append(tempDeleteSuccess ? "成功" : "失败").append("\n");
                                    } else {
                                        detailLog.append("上传文件失败: ").append(fileName).append("\n");
                                    }
                                } else {
                                    detailLog.append("下载文件失败: ").append(fileName).append("\n");
                                }
                            }
                        }
                        
                        // Delete temp directory
                        new File(tempDir).delete();
                        
                    } catch (Exception e) {
                        detailLog.append("连接目标FTP服务器失败: ").append(e.getMessage()).append("\n");
                        throw e;
                    } finally {
                        com.sync.platform.util.FTPUtil.disconnect(targetFtp);
                    }
                } catch (Exception e) {
                    detailLog.append("连接源FTP服务器失败: ").append(e.getMessage()).append("\n");
                    throw e;
                } finally {
                    com.sync.platform.util.FTPUtil.disconnect(sourceFtp);
                }
            } else {
                // Use SSH/SFTP for file transfer (existing implementation)
                detailLog.append("使用SSH/SFTP协议传输文件\n");
                
                // Create SSH session for source server
                com.jcraft.jsch.Session sourceSession = null;
                try {
                    sourceSession = com.sync.platform.util.SSHUtil.createSession(
                        sourceServer.getHost(),
                        sourceServer.getPort(),
                        sourceServer.getUsername(),
                        sourceServer.getPassword(),
                        sourceServer.getPrivateKeyPath()
                    );
                    sourceSession.connect();
                    
                    detailLog.append("成功连接到源SSH/SFTP服务器\n");
                    
                    // Create SSH session for target server
                    com.jcraft.jsch.Session targetSession = null;
                    try {
                        targetSession = com.sync.platform.util.SSHUtil.createSession(
                            targetServer.getHost(),
                            targetServer.getPort(),
                            targetServer.getUsername(),
                            targetServer.getPassword(),
                            targetServer.getPrivateKeyPath()
                        );
                        targetSession.connect();
                        
                        detailLog.append("成功连接到目标SSH/SFTP服务器\n");
                        
                        // List files on source server
                        com.jcraft.jsch.ChannelSftp sourceChannel = (com.jcraft.jsch.ChannelSftp) sourceSession.openChannel("sftp");
                        sourceChannel.connect();
                        
                        @SuppressWarnings("unchecked")
                        Vector<com.jcraft.jsch.ChannelSftp.LsEntry> files = sourceChannel.ls(sourcePath);
                        detailLog.append("源目录文件数: ").append(files.size()).append("\n");
                        
                        // Create temporary directory for file transfer
                        String tempDir = System.getProperty("java.io.tmpdir") + "/sftp_sync_" + task.getId() + "_" + System.currentTimeMillis();
                        new File(tempDir).mkdirs();
                        detailLog.append("创建临时目录: ").append(tempDir).append("\n");
                        
                        // Create target SFTP channel
                        com.jcraft.jsch.ChannelSftp targetChannel = (com.jcraft.jsch.ChannelSftp) targetSession.openChannel("sftp");
                        targetChannel.connect();
                        
                        // Create target directory if it doesn't exist
                        try {
                            com.sync.platform.util.SSHUtil.createRemoteDirectories(targetChannel, targetPath);
                        } catch (Exception e) {
                            detailLog.append("创建目标目录失败: ").append(e.getMessage()).append("\n");
                        }
                        
                        // Process file patterns
                        String[] patterns = com.sync.platform.util.FilePatternUtil.parsePatterns(filePattern);
                        if (patterns != null) {
                            detailLog.append("文件过滤模式: ").append(filePattern).append("\n");
                            for (String pattern : patterns) {
                                detailLog.append("过滤模式: ").append(com.sync.platform.util.FilePatternUtil.getPatternDescription(pattern)).append("\n");
                            }
                        } else {
                            detailLog.append("没有设置文件过滤模式，将同步所有文件\n");
                        }
                        
                        // Process each file
                        for (com.jcraft.jsch.ChannelSftp.LsEntry entry : files) {
                            String fileName = entry.getFilename();
                            boolean isDirectory = entry.getAttrs().isDir();
                            
                            // Skip . and .. entries
                            if (".".equals(fileName) || "..".equals(fileName)) {
                                detailLog.append("跳过特殊目录项: ").append(fileName).append("\n");
                                continue;
                            }
                            
                            detailLog.append("处理项目: ").append(fileName).append(", 类型: ").append(isDirectory ? "目录" : "文件").append("\n");
                            
                            if (isDirectory) {
                                // Process directory - create it but don't filter directories by pattern
                                String sourceSubDir = sourcePath + (sourcePath.endsWith("/") ? "" : "/") + fileName;
                                String targetSubDir = targetPath + (targetPath.endsWith("/") ? "" : "/") + fileName;
                                
                                // Skip directory if not syncing directories
                                if (!syncDirectories) {
                                    detailLog.append("跳过目录 ").append(fileName).append(" (未启用目录同步)\n");
                                    continue;
                                }
                                
                                detailLog.append("创建目标子目录: ").append(targetSubDir).append("\n");
                                
                                try {
                                    // Create target directory
                                    targetChannel.mkdir(targetSubDir);
                                    syncedFiles.add(fileName + "/");
                                    syncCount++;
                                    detailLog.append("目录 ").append(fileName).append(" 同步成功\n");
                                } catch (Exception e) {
                                    detailLog.append("创建目标子目录失败: ").append(e.getMessage()).append("\n");
                                }
                            } else {
                                // Process file
                                // Check if file matches pattern
                                if (patterns != null) {
                                    boolean matched = com.sync.platform.util.FilePatternUtil.matchAnyPattern(fileName, patterns);
                                    if (!matched) {
                                        detailLog.append("文件 ").append(fileName).append(" 不匹配任何过滤模式，跳过\n");
                                        continue;
                                    } else {
                                        detailLog.append("文件 ").append(fileName).append(" 匹配过滤模式\n");
                                    }
                                }
                                
                                String sourceFilePath = sourcePath + (sourcePath.endsWith("/") ? "" : "/") + fileName;
                                String targetFilePath = targetPath + (targetPath.endsWith("/") ? "" : "/") + fileName;
                                String tempFilePath = tempDir + "/" + fileName;
                                
                                detailLog.append("源文件路径: ").append(sourceFilePath).append("\n");
                                detailLog.append("目标文件路径: ").append(targetFilePath).append("\n");
                                detailLog.append("临时文件路径: ").append(tempFilePath).append("\n");
                                
                                try {
                                    // Download file from source
                                    sourceChannel.get(sourceFilePath, tempFilePath);
                                    detailLog.append("下载文件成功\n");
                                    
                                    // Upload file to target
                                    targetChannel.put(tempFilePath, targetFilePath);
                                    detailLog.append("上传文件成功\n");
                                    
                                    syncedFiles.add(fileName);
                                    syncCount++;
                                    detailLog.append("文件 ").append(fileName).append(" 同步成功\n");
                                    
                                    // Delete source file if required
                                    if (deleteSourceFiles) {
                                        try {
                                            sourceChannel.rm(sourceFilePath);
                                            detailLog.append("删除源文件成功\n");
                                        } catch (Exception ex) {
                                            detailLog.append("删除源文件失败: ").append(ex.getMessage()).append("\n");
                                        }
                                    }
                                    
                                    // Delete temp file
                                    boolean tempDeleteSuccess = new File(tempFilePath).delete();
                                    detailLog.append("删除临时文件结果: ").append(tempDeleteSuccess ? "成功" : "失败").append("\n");
                                } catch (Exception e) {
                                    detailLog.append("传输文件失败: ").append(fileName).append(" - ").append(e.getMessage()).append("\n");
                                }
                            }
                        }
                        
                        // Close channels
                        sourceChannel.disconnect();
                        targetChannel.disconnect();
                        
                        // Delete temp directory
                        new File(tempDir).delete();
                        
                    } catch (Exception e) {
                        detailLog.append("连接目标SSH/SFTP服务器失败: ").append(e.getMessage()).append("\n");
                        throw e;
                    } finally {
                        if (targetSession != null) {
                            targetSession.disconnect();
                        }
                    }
                } catch (Exception e) {
                    detailLog.append("连接源SSH/SFTP服务器失败: ").append(e.getMessage()).append("\n");
                    throw e;
                } finally {
                    if (sourceSession != null) {
                        sourceSession.disconnect();
                    }
                }
            }
            
            // Build result
            detailLog.append("\n文件同步成功完成\n");
            detailLog.append("共同步文件: ").append(syncCount).append(" 个\n");
            
            if (!syncedFiles.isEmpty()) {
                detailLog.append("\n同步的文件列表:\n");
                for (String file : syncedFiles) {
                    detailLog.append("- ").append(file).append("\n");
                }
            }
            
            String message = String.format("文件同步成功，共传输 %d 个文件", syncCount);
            result.put("success", true);
            result.put("message", message);
            result.put("syncCount", syncCount);
            result.put("syncedFiles", syncedFiles);
            
            if (logId != null) {
                syncTaskLogService.updateLog(logId, EXECUTION_SUCCESS, syncCount, message, detailLog.toString());
            }
            
            return result;
        } catch (Exception e) {
            String errorMsg = "执行文件同步任务出错: " + e.getMessage();
            logger.error(errorMsg, e);
            
            detailLog.append(errorMsg).append("\n");
            for (StackTraceElement element : e.getStackTrace()) {
                detailLog.append(element.toString()).append("\n");
            }
            
            result.put("message", errorMsg);
            
            if (logId != null) {
                syncTaskLogService.updateLog(logId, EXECUTION_FAILED, 0, errorMsg, detailLog.toString());
            }
            
            return result;
        }
    }
    
    // Helper method to get database connection
    private Connection getConnection(DatabaseConfig dbConfig) {
        try {
            // Ensure URL and driver are set
            if (dbConfig.getUrl() == null || dbConfig.getUrl().isEmpty()) {
                String url = com.sync.platform.util.JdbcUrlBuilder.buildJdbcUrl(
                        dbConfig.getDbType(),
                        dbConfig.getHost(),
                        dbConfig.getPort(),
                        dbConfig.getDatabaseName(),
                        dbConfig.getCharset()
                );
                dbConfig.setUrl(url);
            }

            if (dbConfig.getDriverClass() == null || dbConfig.getDriverClass().isEmpty()) {
                String driverClass = com.sync.platform.util.JdbcUrlBuilder.getDriverClassName(dbConfig.getDbType());
                dbConfig.setDriverClass(driverClass);
            }

            // Get connection
            return com.sync.platform.util.DatabaseUtil.getConnection(
                    dbConfig.getUrl(),
                    dbConfig.getDriverClass(),
                    dbConfig.getUsername(),
                    dbConfig.getPassword()
            );
        } catch (Exception e) {
            logger.error("Error connecting to database {}: {}", dbConfig.getName(), e.getMessage(), e);
            return null;
        }
    }
    
    // Sync a single table
    private int syncTable(Connection sourceConn, Connection targetConn, String tableName, 
                          boolean createTableIfNotExists, boolean truncateBeforeSync,
                          boolean incrementalSync, String incrementalColumn, LocalDateTime lastSyncTime) throws SQLException {
        // Get database types for source and target
        String sourceDbType = sourceConn.getMetaData().getDatabaseProductName().toLowerCase();
        String targetDbType = targetConn.getMetaData().getDatabaseProductName().toLowerCase();
        
        logger.info("Syncing table {} from {} to {} ({})", tableName, sourceDbType, targetDbType, 
                   incrementalSync ? "incremental" : "full");
        
        // Get table structure from source
        List<Map<String, String>> columns = getTableColumns(sourceConn, tableName);
        if (columns == null || columns.isEmpty()) {
            logger.error("Could not get columns for table {}", tableName);
            return -1;
        }
        
        // Check incremental column exists if incremental sync is enabled
        boolean incrementalColumnExists = false;
        if (incrementalSync && incrementalColumn != null && !incrementalColumn.isEmpty()) {
            for (Map<String, String> column : columns) {
                if (column.get("name").equalsIgnoreCase(incrementalColumn)) {
                    incrementalColumnExists = true;
                    break;
                }
            }
            
            if (!incrementalColumnExists) {
                logger.warn("Incremental column '{}' not found in table {}. Falling back to full sync.", 
                           incrementalColumn, tableName);
                incrementalSync = false;
            }
        } else {
            incrementalSync = false; // Disable incremental sync if no column specified
        }
        
        // Check if table exists in target
        boolean tableExists = tableExists(targetConn, tableName);
        
        // Create table if it doesn't exist and createTableIfNotExists is true
        if (!tableExists) {
            if (createTableIfNotExists) {
                logger.info("Creating table {} in target database", tableName);
                try {
                    createTable(targetConn, tableName, columns);
                } catch (SQLException e) {
                    logger.error("Failed to create table {}: {}", tableName, e.getMessage());
                    // Check if table now exists despite error (might have been created with warnings)
                    if (!tableExists(targetConn, tableName)) {
                        return -1;
                    }
                }
            } else {
                logger.error("Table {} does not exist in target database and createTableIfNotExists is false", tableName);
                return -1;
            }
        } else if (tableExists && createTableIfNotExists) {
            // Table exists but we should validate and potentially alter the schema
            try {
                List<Map<String, String>> targetColumns = getTableColumns(targetConn, tableName);
                if (needsSchemaUpdate(columns, targetColumns)) {
                    updateTableSchema(targetConn, tableName, columns, targetColumns, targetDbType);
                }
            } catch (SQLException e) {
                logger.warn("Could not validate table schema for {}: {}", tableName, e.getMessage());
                // Continue anyway - we'll try to insert data
            }
        }
        
        // Truncate target table if truncateBeforeSync is true and we're not doing incremental sync
        if (truncateBeforeSync && !incrementalSync) {
            logger.info("Truncating table {} in target database", tableName);
            try {
                truncateTable(targetConn, tableName);
            } catch (SQLException e) {
                logger.warn("Could not truncate table {}: {}. Attempting to continue with DELETE instead.", 
                           tableName, e.getMessage());
                // Fallback to DELETE if TRUNCATE fails
                try (Statement stmt = targetConn.createStatement()) {
                    stmt.execute("DELETE FROM " + tableName);
                } catch (SQLException ex) {
                    logger.error("Failed to delete data from table {}: {}", tableName, ex.getMessage());
                    // Continue anyway - we might be doing an incremental sync
                }
            }
        }
        
        // Build the WHERE clause for incremental sync
        String whereClause = "";
        if (incrementalSync && incrementalColumnExists && lastSyncTime != null) {
            String formattedTimestamp = getFormattedTimestamp(lastSyncTime, sourceDbType);
            whereClause = " WHERE " + incrementalColumn + " >= " + formattedTimestamp;
            logger.info("Incremental sync using column {} with timestamp {}", incrementalColumn, formattedTimestamp);
        }
        
        // Get data from source table
        String sourceQuery = buildSelectQuery(tableName, columns) + whereClause;
        logger.info("Source query: {}", sourceQuery);
        
        try (Statement sourceStmt = sourceConn.createStatement();
             ResultSet rs = sourceStmt.executeQuery(sourceQuery)) {
            
            // If doing incremental sync and target table exists, we need to use MERGE/UPSERT instead of just INSERT
            // Different databases have different syntax for MERGE/UPSERT
            String insertQuery;
            if (incrementalSync && tableExists) {
                insertQuery = buildUpsertQuery(tableName, columns, targetDbType);
            } else {
                insertQuery = buildInsertQuery(tableName, columns);
            }
            
            // Insert data into target
            int batchSize = 1000; // Process in batches
            int totalRows = 0;
            int batchCount = 0;
            List<String> failedRows = new ArrayList<>();
            
            try (PreparedStatement targetStmt = targetConn.prepareStatement(insertQuery)) {
                while (rs.next()) {
                    batchCount++;
                    totalRows++;
                    
                    try {
                        // Set parameters with type conversion
                        for (int i = 0; i < columns.size(); i++) {
                            Map<String, String> column = columns.get(i);
                            String columnName = column.get("name");
                            Object value = rs.getObject(columnName);
                            
                            // Handle null values
                            if (value == null) {
                                targetStmt.setNull(i + 1, rs.getMetaData().getColumnType(i + 1));
                                continue;
                            }
                            
                            // Handle type conversions between different database systems
                            value = convertValueForTarget(value, column.get("type"), targetDbType);
                            
                            targetStmt.setObject(i + 1, value);
                        }
                        
                        // For UPSERT/MERGE operations in some databases, we need to set parameters twice
                        if (incrementalSync && (targetDbType.contains("postgresql") || targetDbType.contains("oracle"))) {
                            for (int i = 0; i < columns.size(); i++) {
                                Map<String, String> column = columns.get(i);
                                String columnName = column.get("name");
                                Object value = rs.getObject(columnName);
                                
                                // Skip primary keys for the UPDATE part
                                if (column.get("key") != null && column.get("key").equalsIgnoreCase("PRI")) {
                                    continue;
                                }
                                
                                // Handle null values
                                if (value == null) {
                                    targetStmt.setNull(i + columns.size() + 1, rs.getMetaData().getColumnType(i + 1));
                                    continue;
                                }
                                
                                // Handle type conversions between different database systems
                                value = convertValueForTarget(value, column.get("type"), targetDbType);
                                
                                targetStmt.setObject(i + columns.size() + 1, value);
                            }
                        }
                        
                        targetStmt.addBatch();
                    } catch (SQLException e) {
                        // Log the problematic row but continue with others
                        logger.warn("Error preparing row {} for table {}: {}", totalRows, tableName, e.getMessage());
                        failedRows.add("Row " + totalRows + ": " + e.getMessage());
                    }
                    
                    // Execute batch when size reaches limit
                    if (batchCount >= batchSize) {
                        try {
                            targetStmt.executeBatch();
                        } catch (SQLException e) {
                            logger.error("Error executing batch insert for table {}: {}", tableName, e.getMessage());
                            throw e; // Re-throw to trigger rollback
                        }
                        batchCount = 0;
                    }
                }
                
                // Execute remaining batch
                if (batchCount > 0) {
                    try {
                        targetStmt.executeBatch();
                    } catch (SQLException e) {
                        logger.error("Error executing final batch insert for table {}: {}", tableName, e.getMessage());
                        throw e; // Re-throw to trigger rollback
                    }
                }
            }
            
            // Log any failed rows
            if (!failedRows.isEmpty()) {
                logger.warn("Failed to process {} rows for table {}", failedRows.size(), tableName);
            }
            
            return totalRows - failedRows.size();
        }
    }
    
    // Helper method to format timestamp based on database type
    private String getFormattedTimestamp(LocalDateTime timestamp, String dbType) {
        if (dbType.contains("mysql") || dbType.contains("mariadb")) {
            return "'" + timestamp.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "'";
        } else if (dbType.contains("postgresql")) {
            return "'" + timestamp.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "'::timestamp";
        } else if (dbType.contains("oracle")) {
            return "TO_TIMESTAMP('" + timestamp.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "', 'YYYY-MM-DD HH24:MI:SS')";
        } else if (dbType.contains("sqlserver") || dbType.contains("microsoft")) {
            return "'" + timestamp.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "'";
        } else {
            // Default format
            return "'" + timestamp.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "'";
        }
    }
    
    // Get table columns
    private List<Map<String, String>> getTableColumns(Connection conn, String tableName) {
        try {
            return com.sync.platform.util.DatabaseUtil.getColumns(conn, tableName);
        } catch (SQLException e) {
            logger.error("Error getting columns for table {}: {}", tableName, e.getMessage(), e);
            return null;
        }
    }
    
    // Check if table exists
    private boolean tableExists(Connection conn, String tableName) {
        try {
            List<String> tables = com.sync.platform.util.DatabaseUtil.getTables(conn);
            return tables.contains(tableName);
        } catch (SQLException e) {
            logger.error("Error checking if table {} exists: {}", tableName, e.getMessage(), e);
            return false;
        }
    }
    
    // Create table in target database
    private void createTable(Connection conn, String tableName, List<Map<String, String>> columns) throws SQLException {
        // Get database type from connection metadata
        String dbProduct = conn.getMetaData().getDatabaseProductName().toLowerCase();
        
        StringBuilder createTableSQL = new StringBuilder();
        createTableSQL.append("CREATE TABLE ");
        
        // In MySQL we can use IF NOT EXISTS
        if (dbProduct.contains("mysql") || dbProduct.contains("mariadb")) {
            createTableSQL.append("IF NOT EXISTS ");
        }
        
        createTableSQL.append(tableName).append(" (");
        
        // Add primary key identification
        List<String> primaryKeys = getPrimaryKeys(conn, tableName);
        
        for (int i = 0; i < columns.size(); i++) {
            Map<String, String> column = columns.get(i);
            String columnName = column.get("name");
            String columnType = adaptDataType(column.get("type"), dbProduct);
            
            createTableSQL.append(columnName).append(" ").append(columnType);
            
            // Add NOT NULL if column is not nullable
            if (column.get("nullable").equalsIgnoreCase("NO")) {
                createTableSQL.append(" NOT NULL");
            }
            
            // Add PRIMARY KEY directly for SQLite
            if (dbProduct.contains("sqlite") && primaryKeys.contains(columnName)) {
                createTableSQL.append(" PRIMARY KEY");
            }
            
            if (i < columns.size() - 1) {
                createTableSQL.append(", ");
            }
        }
        
        // Add primary key constraint for databases other than SQLite
        if (!dbProduct.contains("sqlite") && !primaryKeys.isEmpty()) {
            createTableSQL.append(", PRIMARY KEY (");
            for (int i = 0; i < primaryKeys.size(); i++) {
                createTableSQL.append(primaryKeys.get(i));
                if (i < primaryKeys.size() - 1) {
                    createTableSQL.append(", ");
                }
            }
            createTableSQL.append(")");
        }
        
        createTableSQL.append(")");
        
        // Add MySQL-specific options
        if (dbProduct.contains("mysql") || dbProduct.contains("mariadb")) {
            createTableSQL.append(" ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci");
        }
        
        logger.info("Creating table with SQL: {}", createTableSQL);
        
        try (Statement stmt = conn.createStatement()) {
            stmt.execute(createTableSQL.toString());
        }
    }
    
    // Get primary keys for a table
    private List<String> getPrimaryKeys(Connection conn, String tableName) {
        List<String> primaryKeys = new ArrayList<>();
        try {
            ResultSet rs = conn.getMetaData().getPrimaryKeys(null, null, tableName);
            while (rs.next()) {
                primaryKeys.add(rs.getString("COLUMN_NAME"));
            }
            rs.close();
        } catch (SQLException e) {
            logger.error("Error getting primary keys for table {}: {}", tableName, e.getMessage(), e);
        }
        return primaryKeys;
    }
    
    // Adapt data types between different database systems
    private String adaptDataType(String sourceType, String targetDbProduct) {
        if (sourceType == null || sourceType.isEmpty()) {
            return "VARCHAR(255)"; // Default fallback
        }
        
        // Normalize source type and extract size info if present
        String baseType = sourceType.split("\\(")[0].trim().toLowerCase();
        String size = "";
        if (sourceType.contains("(") && sourceType.contains(")")) {
            size = sourceType.substring(sourceType.indexOf("("), sourceType.indexOf(")") + 1);
        }
        
        // PostgreSQL target
        if (targetDbProduct.contains("postgresql")) {
            switch (baseType) {
                case "int":
                case "integer":
                case "smallint":
                case "tinyint":
                    if (baseType.equals("tinyint") && size.contains("1")) {
                        return "BOOLEAN";
                    }
                    return size.isEmpty() ? "INTEGER" : "INTEGER";
                case "bigint":
                    return "BIGINT";
                case "decimal":
                case "numeric":
                    return size.isEmpty() ? "NUMERIC(19,6)" : "NUMERIC" + size;
                case "float":
                case "double":
                case "real":
                    return "DOUBLE PRECISION";
                case "bit":
                case "boolean":
                case "bool":
                    return "BOOLEAN";
                case "datetime":
                case "timestamp":
                    return "TIMESTAMP";
                case "date":
                    return "DATE";
                case "time":
                    return "TIME";
                case "varbinary":
                case "binary":
                case "blob":
                case "longblob":
                case "mediumblob":
                    return "BYTEA";
                case "varchar":
                case "nvarchar":
                case "char":
                case "nchar":
                    return size.isEmpty() ? "VARCHAR(255)" : "VARCHAR" + size;
                case "text":
                case "longtext":
                case "mediumtext":
                    return "TEXT";
                default:
                    return "VARCHAR(255)";
            }
        }
        // MySQL target
        else if (targetDbProduct.contains("mysql") || targetDbProduct.contains("mariadb")) {
            switch (baseType) {
                case "int":
                case "integer":
                    return size.isEmpty() ? "INT" : "INT" + size;
                case "smallint":
                    return size.isEmpty() ? "SMALLINT" : "SMALLINT" + size;
                case "tinyint":
                    return size.isEmpty() ? "TINYINT" : "TINYINT" + size;
                case "bigint":
                    return size.isEmpty() ? "BIGINT" : "BIGINT" + size;
                case "decimal":
                case "numeric":
                    return size.isEmpty() ? "DECIMAL(19,6)" : "DECIMAL" + size;
                case "float":
                    return size.isEmpty() ? "FLOAT" : "FLOAT" + size;
                case "double":
                case "real":
                    return size.isEmpty() ? "DOUBLE" : "DOUBLE" + size;
                case "bit":
                case "boolean":
                case "bool":
                    return "TINYINT(1)";
                case "datetime":
                case "timestamp":
                    return "DATETIME";
                case "date":
                    return "DATE";
                case "time":
                    return "TIME";
                case "varbinary":
                case "binary":
                    return size.isEmpty() ? "VARBINARY(255)" : "VARBINARY" + size;
                case "blob":
                    return "BLOB";
                case "longblob":
                    return "LONGBLOB";
                case "mediumblob":
                    return "MEDIUMBLOB";
                case "varchar":
                case "nvarchar":
                    return size.isEmpty() ? "VARCHAR(255)" : "VARCHAR" + size;
                case "char":
                case "nchar":
                    return size.isEmpty() ? "CHAR(1)" : "CHAR" + size;
                case "text":
                    return "TEXT";
                case "longtext":
                    return "LONGTEXT";
                case "mediumtext":
                    return "MEDIUMTEXT";
                default:
                    return "VARCHAR(255)";
            }
        }
        // SQLite target
        else if (targetDbProduct.contains("sqlite")) {
            switch (baseType) {
                case "int":
                case "integer":
                case "smallint":
                case "tinyint":
                case "bigint":
                    return "INTEGER";
                case "decimal":
                case "numeric":
                case "float":
                case "double":
                case "real":
                    return "REAL";
                case "bit":
                case "boolean":
                case "bool":
                    return "INTEGER"; // SQLite doesn't have a true boolean type
                case "datetime":
                case "timestamp":
                case "date":
                case "time":
                    return "TEXT"; // SQLite stores dates as strings in ISO format
                case "varbinary":
                case "binary":
                case "blob":
                case "longblob":
                case "mediumblob":
                    return "BLOB";
                case "varchar":
                case "nvarchar":
                case "char":
                case "nchar":
                case "text":
                case "longtext":
                case "mediumtext":
                    return "TEXT";
                default:
                    return "TEXT";
            }
        }
        // Oracle target
        else if (targetDbProduct.contains("oracle")) {
            switch (baseType) {
                case "int":
                case "integer":
                case "smallint":
                case "tinyint":
                    return "NUMBER(10)";
                case "bigint":
                    return "NUMBER(19)";
                case "decimal":
                case "numeric":
                    return size.isEmpty() ? "NUMBER(19,6)" : "NUMBER" + size;
                case "float":
                    return "BINARY_FLOAT";
                case "double":
                case "real":
                    return "BINARY_DOUBLE";
                case "bit":
                case "boolean":
                case "bool":
                    return "NUMBER(1)";
                case "datetime":
                case "timestamp":
                    return "TIMESTAMP";
                case "date":
                    return "DATE";
                case "time":
                    return "TIMESTAMP";
                case "varbinary":
                case "binary":
                case "blob":
                case "longblob":
                case "mediumblob":
                    return "BLOB";
                case "varchar":
                case "nvarchar":
                    return size.isEmpty() ? "VARCHAR2(255)" : "VARCHAR2" + size;
                case "char":
                case "nchar":
                    return size.isEmpty() ? "CHAR(1)" : "CHAR" + size;
                case "text":
                case "longtext":
                case "mediumtext":
                    return "CLOB";
                default:
                    return "VARCHAR2(255)";
            }
        }
        // SQL Server target
        else if (targetDbProduct.contains("microsoft") || targetDbProduct.contains("sqlserver")) {
            switch (baseType) {
                case "int":
                case "integer":
                    return "INT";
                case "smallint":
                    return "SMALLINT";
                case "tinyint":
                    return "TINYINT";
                case "bigint":
                    return "BIGINT";
                case "decimal":
                case "numeric":
                    return size.isEmpty() ? "DECIMAL(19,6)" : "DECIMAL" + size;
                case "float":
                case "double":
                case "real":
                    return "FLOAT";
                case "bit":
                case "boolean":
                case "bool":
                    return "BIT";
                case "datetime":
                case "timestamp":
                    return "DATETIME2";
                case "date":
                    return "DATE";
                case "time":
                    return "TIME";
                case "varbinary":
                case "binary":
                    return size.isEmpty() ? "VARBINARY(MAX)" : "VARBINARY" + size;
                case "blob":
                case "longblob":
                case "mediumblob":
                    return "VARBINARY(MAX)";
                case "varchar":
                    return size.isEmpty() ? "VARCHAR(255)" : "VARCHAR" + size;
                case "nvarchar":
                    return size.isEmpty() ? "NVARCHAR(255)" : "NVARCHAR" + size;
                case "char":
                    return size.isEmpty() ? "CHAR(1)" : "CHAR" + size;
                case "nchar":
                    return size.isEmpty() ? "NCHAR(1)" : "NCHAR" + size;
                case "text":
                    return "VARCHAR(MAX)";
                case "longtext":
                case "mediumtext":
                    return "VARCHAR(MAX)";
                default:
                    return "VARCHAR(255)";
            }
        }
        
        // Default fallback - return the original type
        return sourceType;
    }
    
    // Truncate table
    private void truncateTable(Connection conn, String tableName) throws SQLException {
        // Get database type
        String dbProduct = conn.getMetaData().getDatabaseProductName().toLowerCase();
        
        String truncateSQL;
        if (dbProduct.contains("mysql") || dbProduct.contains("mariadb") || 
            dbProduct.contains("postgresql")) {
            // MySQL and PostgreSQL support TRUNCATE
            truncateSQL = "TRUNCATE TABLE " + tableName;
        } else {
            // Other databases like SQLite don't support TRUNCATE
            truncateSQL = "DELETE FROM " + tableName;
        }
        
        try (Statement stmt = conn.createStatement()) {
            stmt.execute(truncateSQL);
        }
    }
    
    // Build SELECT query
    private String buildSelectQuery(String tableName, List<Map<String, String>> columns) {
        StringBuilder selectSQL = new StringBuilder();
        selectSQL.append("SELECT ");
        
        for (int i = 0; i < columns.size(); i++) {
            selectSQL.append(columns.get(i).get("name"));
            if (i < columns.size() - 1) {
                selectSQL.append(", ");
            }
        }
        
        selectSQL.append(" FROM ").append(tableName);
        return selectSQL.toString();
    }
    
    // Build INSERT query
    private String buildInsertQuery(String tableName, List<Map<String, String>> columns) {
        StringBuilder insertSQL = new StringBuilder();
        insertSQL.append("INSERT INTO ").append(tableName).append(" (");
        
        for (int i = 0; i < columns.size(); i++) {
            insertSQL.append(columns.get(i).get("name"));
            if (i < columns.size() - 1) {
                insertSQL.append(", ");
            }
        }
        
        insertSQL.append(") VALUES (");
        
        for (int i = 0; i < columns.size(); i++) {
            insertSQL.append("?");
            if (i < columns.size() - 1) {
                insertSQL.append(", ");
            }
        }
        
        insertSQL.append(")");
        return insertSQL.toString();
    }

    @Override
    public boolean updateTaskStatus(Long id, Integer status) {
        return lambdaUpdate()
                .eq(SyncTask::getId, id)
                .set(SyncTask::getStatus, status)
                .update();
    }
    
    @Override
    public int countDatabaseSyncTasks() {
        return Math.toIntExact(lambdaQuery()
                .eq(SyncTask::getTaskType, 1) // Database sync type
                .count());
    }
    
    @Override
    public int countFileSyncTasks() {
        return Math.toIntExact(lambdaQuery()
                .eq(SyncTask::getTaskType, 2) // File sync type
                .count());
    }

    // Check if target table schema needs updating
    private boolean needsSchemaUpdate(List<Map<String, String>> sourceColumns, 
                                     List<Map<String, String>> targetColumns) {
        // If column count differs, schema needs update
        if (sourceColumns.size() != targetColumns.size()) {
            return true;
        }
        
        // Create maps of column names to columns for easier comparison
        Map<String, Map<String, String>> sourceColumnMap = new HashMap<>();
        Map<String, Map<String, String>> targetColumnMap = new HashMap<>();
        
        for (Map<String, String> column : sourceColumns) {
            sourceColumnMap.put(column.get("name").toLowerCase(), column);
        }
        
        for (Map<String, String> column : targetColumns) {
            targetColumnMap.put(column.get("name").toLowerCase(), column);
        }
        
        // Check if all source columns exist in target
        for (String columnName : sourceColumnMap.keySet()) {
            if (!targetColumnMap.containsKey(columnName)) {
                return true; // Missing column
            }
        }
        
        return false;
    }
    
    // Update table schema to match source
    private void updateTableSchema(Connection conn, String tableName, 
                                  List<Map<String, String>> sourceColumns,
                                  List<Map<String, String>> targetColumns,
                                  String dbType) throws SQLException {
        // Create maps of column names to columns
        Map<String, Map<String, String>> sourceColumnMap = new HashMap<>();
        Map<String, Map<String, String>> targetColumnMap = new HashMap<>();
        
        for (Map<String, String> column : sourceColumns) {
            sourceColumnMap.put(column.get("name").toLowerCase(), column);
        }
        
        for (Map<String, String> column : targetColumns) {
            targetColumnMap.put(column.get("name").toLowerCase(), column);
        }
        
        // Find columns to add
        List<Map<String, String>> columnsToAdd = new ArrayList<>();
        for (String columnName : sourceColumnMap.keySet()) {
            if (!targetColumnMap.containsKey(columnName)) {
                columnsToAdd.add(sourceColumnMap.get(columnName));
            }
        }
        
        // Add missing columns
        if (!columnsToAdd.isEmpty()) {
            for (Map<String, String> column : columnsToAdd) {
                String columnName = column.get("name");
                String columnType = adaptDataType(column.get("type"), dbType);
                String nullable = column.get("nullable").equalsIgnoreCase("NO") ? " NOT NULL" : "";
                
                String alterSql = "ALTER TABLE " + tableName + " ADD COLUMN " + columnName + " " + columnType + nullable;
                
                logger.info("Adding column with SQL: {}", alterSql);
                try (Statement stmt = conn.createStatement()) {
                    stmt.execute(alterSql);
                    logger.info("Added column {} to table {}", columnName, tableName);
                } catch (SQLException e) {
                    logger.warn("Could not add column {} to table {}: {}", columnName, tableName, e.getMessage());
                    // Continue with other columns even if one fails
                }
            }
        }
    }
    
    // Convert a value for the target database
    private Object convertValueForTarget(Object value, String sourceType, String targetDbType) {
        if (value == null) {
            return null;
        }
        
        // Normalize source type
        String baseSourceType = sourceType.split("\\(")[0].trim().toLowerCase();
        
        // Handle boolean conversions
        if (baseSourceType.equals("bit") || baseSourceType.equals("boolean") || baseSourceType.equals("bool") || 
            (baseSourceType.equals("tinyint") && sourceType.contains("(1)"))) {
            
            // Convert to appropriate boolean representation based on target database
            if (targetDbType.contains("postgresql")) {
                if (value instanceof Number) {
                    return ((Number) value).intValue() != 0;
                } else if (value instanceof String) {
                    String strVal = ((String) value).toLowerCase();
                    return "1".equals(strVal) || "true".equals(strVal) || "t".equals(strVal) || "yes".equals(strVal) || "y".equals(strVal);
                }
                return value;
            } else if (targetDbType.contains("mysql") || targetDbType.contains("mariadb")) {
                if (value instanceof Boolean) {
                    return (Boolean) value ? (byte) 1 : (byte) 0;
                } else if (value instanceof String) {
                    String strVal = ((String) value).toLowerCase();
                    return "true".equals(strVal) || "t".equals(strVal) || "yes".equals(strVal) || "y".equals(strVal) || "1".equals(strVal) ? (byte) 1 : (byte) 0;
                }
                return value;
            } else if (targetDbType.contains("oracle")) {
                if (value instanceof Boolean) {
                    return (Boolean) value ? 1 : 0;
                } else if (value instanceof String) {
                    String strVal = ((String) value).toLowerCase();
                    return "true".equals(strVal) || "t".equals(strVal) || "yes".equals(strVal) || "y".equals(strVal) || "1".equals(strVal) ? 1 : 0;
                }
                return value;
            } else if (targetDbType.contains("sqlserver") || targetDbType.contains("microsoft")) {
                if (value instanceof Number) {
                    return ((Number) value).intValue() != 0;
                } else if (value instanceof String) {
                    String strVal = ((String) value).toLowerCase();
                    return "1".equals(strVal) || "true".equals(strVal) || "t".equals(strVal) || "yes".equals(strVal) || "y".equals(strVal);
                }
                return value;
            }
        }
        
        // Handle date/time conversions
        if (baseSourceType.equals("date") || baseSourceType.equals("datetime") || 
            baseSourceType.equals("timestamp") || baseSourceType.equals("time")) {
            
            // Convert to appropriate date representation based on target database type
            if (value instanceof String) {
                try {
                    // Try parsing as ISO date format (most common)
                    SimpleDateFormat isoFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    Date date = isoFormat.parse((String) value);
                    
                    if (targetDbType.contains("sqlite")) {
                        // SQLite stores dates as ISO strings
                        return isoFormat.format(date);
                    } else {
                        // For other databases, return as java.sql.Timestamp
                        return new java.sql.Timestamp(date.getTime());
                    }
                } catch (java.text.ParseException e) {
                    // If parsing fails, return original value
                    logger.error("Failed to parse date value: {}", value);
                    return value;
                }
            } else if (value instanceof Number) {
                // Handle Unix timestamp (numeric timestamp)
                long timestamp = ((Number) value).longValue();
                if (targetDbType.contains("sqlite")) {
                    // Convert to ISO string for SQLite
                    SimpleDateFormat isoFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    return isoFormat.format(new Date(timestamp));
                } else {
                    // For other databases, return as java.sql.Timestamp
                    return new java.sql.Timestamp(timestamp);
                }
            }
        }
        
        // Handle numeric conversions
        if (baseSourceType.equals("decimal") || baseSourceType.equals("numeric") || 
            baseSourceType.equals("float") || baseSourceType.equals("double") || 
            baseSourceType.equals("real") || baseSourceType.equals("money")) {
            
            if (value instanceof String) {
                try {
                    // Try to parse string as BigDecimal for maximum precision
                    return new BigDecimal((String) value);
                } catch (NumberFormatException e) {
                    logger.error("Failed to parse numeric value: {}", value);
                    return value;
                }
            }
        }
        
        // Handle integer conversions
        if (baseSourceType.equals("int") || baseSourceType.equals("integer") || 
            baseSourceType.equals("smallint") || baseSourceType.equals("tinyint") || 
            baseSourceType.equals("bigint")) {
            
            if (value instanceof String) {
                try {
                    String strVal = (String) value;
                    // Use Long for larger values, Integer for smaller ones
                    if (baseSourceType.equals("bigint")) {
                        return Long.parseLong(strVal);
                    } else {
                        return Integer.parseInt(strVal);
                    }
                } catch (NumberFormatException e) {
                    logger.error("Failed to parse integer value: {}", value);
                    return value;
                }
            }
        }
        
        // Handle BLOB/BYTEA conversions
        if (baseSourceType.equals("blob") || baseSourceType.equals("bytea") || 
            baseSourceType.equals("binary") || baseSourceType.equals("varbinary") || 
            baseSourceType.equals("longblob") || baseSourceType.equals("mediumblob")) {
            
            // Ensure proper byte array representation regardless of source format
            if (value instanceof String) {
                // For string representations, encode as UTF-8 bytes
                try {
                    return ((String) value).getBytes("UTF-8");
                } catch (UnsupportedEncodingException e) {
                    logger.error("Failed to convert string to byte array: {}", e.getMessage());
                    return value;
                }
            }
        }
        
        // For other types or if no conversion was needed, return the original value
        return value;
    }

    // Build UPSERT/MERGE query based on database type
    private String buildUpsertQuery(String tableName, List<Map<String, String>> columns, String dbType) {
        // First, find primary key columns
        List<String> primaryKeys = new ArrayList<>();
        for (Map<String, String> column : columns) {
            if (column.get("key") != null && column.get("key").equalsIgnoreCase("PRI")) {
                primaryKeys.add(column.get("name"));
            }
        }
        
        // If no primary keys found, fall back to regular INSERT
        if (primaryKeys.isEmpty()) {
            logger.warn("No primary keys found for table {}. Using regular INSERT instead of UPSERT.", tableName);
            return buildInsertQuery(tableName, columns);
        }
        
        StringBuilder sql = new StringBuilder();
        
        // MySQL / MariaDB - Using INSERT ... ON DUPLICATE KEY UPDATE
        if (dbType.contains("mysql") || dbType.contains("mariadb")) {
            sql.append("INSERT INTO ").append(tableName).append(" (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append(columns.get(i).get("name"));
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(") VALUES (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append("?");
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(") ON DUPLICATE KEY UPDATE ");
            
            boolean firstUpdateColumn = true;
            for (Map<String, String> column : columns) {
                // Don't update primary keys
                if (column.get("key") == null || !column.get("key").equalsIgnoreCase("PRI")) {
                    if (!firstUpdateColumn) {
                        sql.append(", ");
                    }
                    String colName = column.get("name");
                    sql.append(colName).append(" = VALUES(").append(colName).append(")");
                    firstUpdateColumn = false;
                }
            }
            
            return sql.toString();
        }
        
        // PostgreSQL - Using INSERT ... ON CONFLICT ... DO UPDATE
        else if (dbType.contains("postgresql")) {
            sql.append("INSERT INTO ").append(tableName).append(" (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append(columns.get(i).get("name"));
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(") VALUES (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append("?");
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(") ON CONFLICT (");
            
            for (int i = 0; i < primaryKeys.size(); i++) {
                sql.append(primaryKeys.get(i));
                if (i < primaryKeys.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(") DO UPDATE SET ");
            
            boolean firstUpdateColumn = true;
            for (Map<String, String> column : columns) {
                // Don't update primary keys
                if (column.get("key") == null || !column.get("key").equalsIgnoreCase("PRI")) {
                    if (!firstUpdateColumn) {
                        sql.append(", ");
                    }
                    String colName = column.get("name");
                    sql.append(colName).append(" = EXCLUDED.").append(colName);
                    firstUpdateColumn = false;
                }
            }
            
            return sql.toString();
        }
        
        // SQLite - Using INSERT OR REPLACE
        else if (dbType.contains("sqlite")) {
            sql.append("INSERT OR REPLACE INTO ").append(tableName).append(" (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append(columns.get(i).get("name"));
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(") VALUES (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append("?");
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(")");
            
            return sql.toString();
        }
        
        // SQL Server - Using MERGE
        else if (dbType.contains("sqlserver") || dbType.contains("microsoft")) {
            String tempTableName = "#temp_" + tableName;
            
            sql.append("DECLARE @tempTable TABLE (");
            
            for (int i = 0; i < columns.size(); i++) {
                Map<String, String> column = columns.get(i);
                sql.append(column.get("name")).append(" ").append(adaptDataType(column.get("type"), dbType));
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append("); ");
            
            sql.append("INSERT INTO @tempTable VALUES (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append("?");
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append("); ");
            
            sql.append("MERGE INTO ").append(tableName).append(" AS target ");
            sql.append("USING @tempTable AS source ON ");
            
            for (int i = 0; i < primaryKeys.size(); i++) {
                if (i > 0) {
                    sql.append(" AND ");
                }
                sql.append("target.").append(primaryKeys.get(i))
                   .append(" = source.").append(primaryKeys.get(i));
            }
            
            sql.append(" WHEN MATCHED THEN UPDATE SET ");
            
            boolean firstUpdateColumn = true;
            for (Map<String, String> column : columns) {
                // Don't update primary keys
                if (column.get("key") == null || !column.get("key").equalsIgnoreCase("PRI")) {
                    if (!firstUpdateColumn) {
                        sql.append(", ");
                    }
                    String colName = column.get("name");
                    sql.append("target.").append(colName).append(" = source.").append(colName);
                    firstUpdateColumn = false;
                }
            }
            
            sql.append(" WHEN NOT MATCHED THEN INSERT (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append(columns.get(i).get("name"));
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(") VALUES (");
            
            for (int i = 0; i < columns.size(); i++) {
                sql.append("source.").append(columns.get(i).get("name"));
                if (i < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            
            sql.append(");");
            
            return sql.toString();
        }
        
        // Default to regular INSERT for other databases
        logger.warn("UPSERT not supported for database type: {}. Using regular INSERT.", dbType);
        return buildInsertQuery(tableName, columns);
    }
}
